package ka

import (
	"context"
	"errors"
	"github.com/Shopify/sarama"
	"github.com/gogf/gf/container/gvar"
	"github.com/gogf/gf/frame/g"
	"go.opentelemetry.io/otel/label"
	"go.opentelemetry.io/otel/trace"
	"sync"
)

const (
	gTestPrefix         = "test_consumer_"
	gMsgDone            = ""
	gProducerTracerName = "git.medlinker.com/golang/ka/github.com/Shopify/sarama/producer"
	gConsumerTracerName = "git.medlinker.com/golang/ka/github.com/Shopify/sarama/consumer"

	gKafkaPartitionKey = label.Key("messaging.kafka.partition")
)

var (
	ver = sarama.V0_10_2_0
)

var (
	ErrConsumerNotFound = errors.New("consumer not found")
)

type (
	HandleErrorFunc   func(error)
	HandleSucceedFunc func(*sarama.ProducerMessage)
	// 如果返回error，则这一条消费不会被mark消费成功
	HandleConsumerFunc func(ctx context.Context, e *Event) error
)

type Producer struct {
	config   *sarama.Config
	producer sarama.AsyncProducer
	e        HandleErrorFunc
	s        HandleSucceedFunc

	tracer trace.Tracer

	metas g.MapAnyAny
}

type Consumer struct {
	run bool

	ctx    context.Context
	cancel context.CancelFunc

	group     sarama.ConsumerGroup
	saramaCfg *sarama.Config
	// 错误的handler
	e  HandleErrorFunc
	wg *sync.WaitGroup

	cfg   *ConsumerConfig
	topic string

	consumerTopicHandler  HandleConsumerFunc
	consumerEventHandlers map[string]HandleConsumerFunc
	handlerLock           sync.RWMutex

	tracer trace.Tracer
}

type Event struct {
	Time     string    `json:"Time,omitempty"`
	Hostname string    `json:"Hostname,omitempty"`
	From     string    `json:"From,omitempty"`
	Type     string    `json:"Type,omitempty"`
	Data     *gvar.Var `json:"Data,omitempty"`
	// meta的存储
	Kv g.MapStrStr `json:"_trace_kv,omitempty"`
}

func (c *Event) Get(key string) string {
	return c.Kv[key]
}

func (c *Event) Set(key, val string) {
	c.Kv[key] = val
}

type eventProducer struct {
	Time     string `json:"Time,omitempty"`
	From     string `json:"From,omitempty"`
	Hostname string `json:"Hostname,omitempty"`
	Type     string `json:"Type,omitempty"`
	Data     []byte `json:"Data,omitempty"`

	// meta的存储
	Kv g.MapStrStr `json:"_trace_kv,omitempty"`
}

func (c *eventProducer) Get(key string) string {
	return c.Kv[key]
}

func (c *eventProducer) Set(key, val string) {
	c.Kv[key] = val
}

func SetVersion(v sarama.KafkaVersion) {
	ver = v
}

func GetVersion() sarama.KafkaVersion {
	return ver
}

type MessageCarrier struct {
	data []byte
	kv   g.MapStrStr
}
